package com.zc.manager.device.redis.stream;

import com.alibaba.fastjson.JSON;
import com.zc.common.core.redis.stream.RedisStream;
import com.zc.component.gateway.annotation.Subscribe;
import com.zc.core.device.DeviceRegistry;
import com.zc.core.message.DeviceMessage;
import com.zc.manager.device.constant.RedisStreamConstants;
import com.zc.manager.device.service.IIotProductService;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.lang.reflect.Type;
import java.util.HashSet;
import java.util.Set;

/**
 * 设备消息发布
 */
//@Component
public class DeviceMessagePublish {

    @Resource
    private RedisStream redisStream;

    @Resource
    private DeviceRegistry registry;

    @Resource
    private IIotProductService iotProductService;

    /**
     * 设备在线消息添加到redis消息队列
     * @param msg 在线消息
     */
    @Subscribe("/device/*/*/online")
    public void onlineMessage(DeviceMessage msg) {
        onlineOfflineMessage(msg, RedisStreamConstants.DeviceOnline.KEY);
    }

    /**
     * 设备离线消息添加到redis消息队列
     * @param msg 离线消息
     */
    @Subscribe("/device/*/*/offline")
    public void offlineMessage(DeviceMessage msg) {
        onlineOfflineMessage(msg, RedisStreamConstants.DeviceOffline.KEY);
    }

    private void onlineOfflineMessage(DeviceMessage msg, String streamGroup) {
        registry
                .getDevice(msg.getDeviceId())
                .map(deviceOperator ->
                        deviceOperator
                                .getProduct()
                                .map(deviceProductOperator -> iotProductService.selectIotProductById(deviceProductOperator.getId())
                                ).flatMap(iotProduct -> {
                                    String deviceType = iotProduct.getDeviceType();
                                    if ("gateway".equals(deviceType)) {
                                        return deviceOperator.getSelfConfig("childrenId") .flatMap(value -> {
                                            Set<String> ids = JSON.parseObject(value.asString(), (Type)Set.class);
                                            return Mono.just(ids);
                                        });
                                    } else if ("device".equals(deviceType)) {
                                        Set<String> ids = new HashSet<>();
                                        ids.add(deviceOperator.getDeviceId());
                                        return Mono.just(ids);
                                    }
                                    return Mono.empty();
                                }).flatMap(ids -> {
                                    if (ids != null && !ids.isEmpty()) {
                                        return Mono.just(redisStream.add(streamGroup, "deviceIds", JSON.toJSONString(ids)));
                                    }
                                    return Mono.empty();
                                })
                )
                .subscribe(Mono::subscribe);
    }


    /**
     * 设备事件消息添加到redis消息队列
     * @param msg 事件消息
     */
    @Subscribe("/device/*/*/message/event/*")
    public void eventMessage(DeviceMessage msg) {
        redisStream.add(RedisStreamConstants.DeviceEvent.KEY, "content", msg.toJson().toJSONString());
    }

    /**
     * 设备属性上报消息添加到redis消息队列
     * @param msg 属性上报消息
     */
    @Subscribe("/device/*/*/message/property/report")
    public void propertyReportMessage(DeviceMessage msg) {
        redisStream.add(RedisStreamConstants.DevicePropertyReport.KEY, "content", msg.toJson().toJSONString());
    }

    /**
     * 设备属性读回复消息添加到redis消息队列
     * @param msg 属性上报消息
     */
    @Subscribe("/device/*/*/message/property/read/reply")
    public void propertyReadReplyMessage(DeviceMessage msg) {
        redisStream.add(RedisStreamConstants.DevicePropertyReadReply.KEY, "content", msg.toJson().toJSONString());
    }

    /**
     * 设备属性写回复消息添加到redis消息队列
     * @param msg 属性上报消息
     */
    @Subscribe("/device/*/*/message/property/write/reply")
    public void propertyWriteReplyMessage(DeviceMessage msg) {
        redisStream.add(RedisStreamConstants.DevicePropertyWriteReply.KEY, "content", msg.toJson().toJSONString());
    }

    /**
     * 设备属性写回复消息添加到redis消息队列
     * @param msg 属性上报消息
     */
    @Subscribe("/device/*/*/message/function/invoke/reply")
    public void functionReplyMessage(DeviceMessage msg) {
        redisStream.add(RedisStreamConstants.DeviceFunctionReply.KEY, "content", msg.toJson().toJSONString());
    }
}
